package mani.multithreading;

import java.util.LinkedList;
import java.util.concurrent.Callable;
@SuppressWarnings("rawtypes")
public class ThreadPoolTest {
	@SuppressWarnings({ "unchecked" })
	public static void main(String[] args) {
		ThreadPool pool = new ThreadPool(5);
		for (int i = 0; i <= 10; i++) {
			pool.submit(new Task("Task" + i));
		}
		pool.stopThreadPool();
	}
}

@SuppressWarnings("rawtypes")
class Task implements Callable {
	private String taskName;

	public Task(String taskName) {
		this.taskName = taskName;
	}

	@Override
	public Object call() throws Exception {
		System.out.println(" " + taskName + " completed by :"
				+ Thread.currentThread().getName());
		Thread.sleep(1000);
		return "";
	}
}

class myThread<T> extends Thread {
	private ThreadPool<T> pool;
	private boolean active = true;

	public boolean isActive() {
		return active;
	}

	public void setPool(ThreadPool<T> pool) {
		this.pool = pool;
	}

	public void run() {
		System.out.println("Thread Created In ThreadPool is "
				+ Thread.currentThread().getName());
		Callable<T> task;
		while (true) {// same thread will continously check for task in Queue
			task = pool.removeTask();
			if (task != null) {
				try {
					task.call();
				} catch (Exception e) {

					e.printStackTrace();
				}
			} else {
				if (!isActive()) {
					break;// to terminate the thread
				} else {
					synchronized (pool.getWaitlock()) {// If no Task will there
														// then pool will wait
														// for task
						try {
							pool.getWaitlock().wait();
						} catch (InterruptedException e) {

							e.printStackTrace();
						}
					}
				}
			}
		}
	}

	public void stopThread() {
		active = false;
	}
}

class ThreadPool<T> {
	private Object waitlock = new Object();
	private final LinkedList<myThread<T>> threads;
	private final LinkedList<Callable<T>> tasks;
	private volatile boolean shutdown;

	public ThreadPool(int size) {
		threads = new LinkedList<myThread<T>>();
		tasks = new LinkedList<Callable<T>>();
		shutdown = false;

		for (int i = 0; i < size; i++) {
			myThread<T> mt = new myThread<T>();
			mt.setPool(this);
			threads.add(mt);
			mt.start();
		}

	}

	public boolean isShutDown() {
		return shutdown;
	}

	public synchronized void addTask(Callable<T> c) {
		tasks.add(c);
	}

	public synchronized Callable<T> removeTask() {
		return tasks.poll();
	}

	public void submit(Callable<T> c) {
		if (!shutdown) {
			addTask(c);
			synchronized (this.waitlock) {// If any thread are waiting due to
											// lack of task then it will notify
											// that thread
				this.waitlock.notify();
			}
		} else {
			System.out
					.println("Thread Pool is shutDown ..Please create the ThreadPool");
		}
	}

	public void stopThreadPool() {
		for (myThread<T> mt : threads) {
			mt.stopThread();// it will terminate the all thread
		}
		synchronized (this.waitlock) {
			this.waitlock.notifyAll();// if some thread are in wait state then
										// it will notify
		}

		for (myThread<T> mt : threads) {
			try {
				mt.join();// it will join the other threads so that if atleast
							// any one in running state it will join all threads
							// in threadpool
			} catch (InterruptedException e) {

				e.printStackTrace();
			}
		}
	}

	public Object getWaitlock() {
		return waitlock;
	}

}
